/**
 *
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 *
 */

package org.jiopi.ibean.module.remote.server;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import org.apache.log4j.Logger;
import org.jiopi.ibean.module.remote.common.ByteUtil;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * process a client's request
 *
 * commands:
 * register ID //register client's
 * -register ID //register server ID to client
 * call {length} //call a method
 *
 *
 * @author netstarry
 * @version 0.1.0.5 2010.10.31
 * @since 0.1.0.5 2010.10.31
 *
 */
public class ClientProcessor implements Runnable{

    private static Logger _logger = Logger.getLogger(ClientProcessor.class);

    protected Socket socket;

    protected final Receiver receiver = new Receiver();

    protected final Sender sender = new Sender();

    public ClientProcessor(Socket socket){
        this.socket = socket;
    }

    public void run() {
        new Thread(sender).start();
        new Thread(receiver).start();
    }

    public synchronized void close(){
        if(this.socket!=null){
            try {
                this.socket.close();
            } catch (IOException e) {
                _logger.warn("",e);
            }
            this.socket=null;
        }
    }

    private class Receiver implements Runnable{

        public void run(){
            if(socket!=null){
                try {
                    InputStream is = socket.getInputStream();
                    byte[] longBytes = new byte[8];
                    byte[] readBytes;
                    int readNum;
                    do{
                        readNum = is.read(longBytes);
                        if(readNum == 8){
                            int packageLength = ByteUtil.readIntBig(longBytes);
                            readBytes = new byte[packageLength];
                            readNum = is.read(readBytes);
                            //TODO replace with real code
                            System.out.println("read:"+new String(readBytes));
                        }
                    }while(readNum>0);
                } catch (IOException e) {
                    _logger.error("",e);
                }
            }
            close();
        }
    }

    private class Sender implements Runnable{
        private BlockingQueue<byte[]> queue = new LinkedBlockingQueue<byte[]>();

        public void send(byte[] content){
            this.queue.add(content);
        }

        public void run(){
            if(socket!=null){
                try {
                    OutputStream os = socket.getOutputStream();

                    byte[] content;

                    do{
                        try {
                            content = this.queue.take();
                            os.write(content);                            
                        } catch (InterruptedException e) {
                            content=null;
                            _logger.warn("",e);
                        }
                    }while(content!=null);

                } catch (IOException e) {
                    _logger.error("",e);
                }
            }
            close();
        }
    }
}
